
--排序函数
create temporary function tab_scan_trace_step as 'com.yunlu.bigdata.scala.function.ScanStep' using jar 'hdfs:///user/jokic.wang/udf/scanstep.jar';

insert overwrite table jms_tmp.tmp_tab_scan_step_detail
select waybill_no,
       exp_array[0] groupbystep,
       exp_array[1] step,
       exp_array[2] stepcenter,
       exp_array[3] scan_type,
       exp_array[4] network_code,
       exp_array[5] network_name,
       exp_array[6] network_type,
       exp_array[7] scan_time,
       exp_array[8] scan_infos,
       date(exp_array[7]) dt
from(
select waybill_no, exp, split(exp, '&') exp_array
from (select waybill_no, concat_ws("|", sort_array(collect_set(scan_infos))) scan_array --根据扫描时间前缀排序
      from (select waybill_no,
                   concat_ws('@',
                             coalesce(a.scan_time, ''),
                             coalesce(a.scan_site_code, ''),
                             coalesce(b.network_type, ''),
                             coalesce(a.scan_site, ''),
                             coalesce(a.data_type, ''),
                             coalesce(a.record_id, ''),
                             coalesce(a.shipment_no, ''),
                             coalesce(a.package_code, ''),
                             coalesce(a.scan_type, ''),
                             coalesce(a.next_station, ''),
                             coalesce(a.next_station_code, ''),
                             coalesce(a.scan_user, ''),
                             coalesce(a.scan_user_code, ''),
                             coalesce(a.scan_user_id, ''),
                             coalesce(a.send_site, ''),
                             coalesce(a.pcs, ''),
                             coalesce(a.weight, ''),
                             coalesce(a.upload_time, '')
                       ) scan_infos
            from (select 1               as data_type
                       , record_id                    --记录编号_ef
                       , waybill_no                   --运单编号_ef
                       , shipment_no                  --任务号
                       , package_code                 --所属包号_ef
                       , scan_type                    --扫描类型_ef
                       , next_station                 --下一站_ef
                       , next_station_code            --下一站编号_ef
                       , scan_user                    --扫描雇员_ef
                       , scan_user_code               --扫描雇员编号_ef
                       , scan_user_id                 --扫描雇员ID_ef
                       , input_site      as scan_site --扫描网点_ef
                       , scan_site_code               --扫描网点编号_ef
                       , send_site                    --寄件网点_ef
                       , scan_time                    --扫描时间_ef
                       , pcs                          --件数
                       , weight                       --重量
                       , upload_time                  --上传时间_ef
                       , date(scan_time) as dt        --扫描时间分区
                  from jms_dwd.dwd_tab_barscan_send_base_dt --全量发件表
                  where dt between date_add('{{ execution_date | cst_ds }}', -4 + 1)
                      and '{{ execution_date | cst_ds }}'
                    and date(scan_time) between date_add('{{ execution_date | cst_ds }}', -4 + 1)
                      and '{{ execution_date | cst_ds }}'
                    and scan_site_code is not null
                  union all
                  select 2               as data_type
                       , record_id                      --记录编号_ef
                       , waybill_no                     --运单编号_ef
                       , transfer_code   as shipment_no --任务号
                       , package_code                   --所属包号_ef
                       , scan_type                      --扫描类型_ef
                       , pre_station                    --上一站_ef
                       , pre_station_code               --上一站编号_ef
                       , scan_user                      --扫描雇员_ef
                       , scan_user_code                 --扫描雇员编号_ef
                       , scan_user_id                   --扫描雇员ID_ef
                       , scan_site                      --扫描网点_ef
                       , scan_site_code                 --扫描网点编号_ef
                       , send_site                      --寄件网点_ef
                       , scan_time                      --扫描时间_ef
                       , pcs                            --件数
                       , weight                         --重量
                       , upload_time                    --上传时间_ef
                       , date(scan_time) as dt          --扫描时间分区
                  from jms_dwd.dwd_tab_barscan_arrival_base_dt --全量到件表
                  where dt between date_add('{{ execution_date | cst_ds }}', -4 + 1)
                      and '{{ execution_date | cst_ds }}'
                    and date(scan_time) between date_add('{{ execution_date | cst_ds }}', -4 + 1)
                      and '{{ execution_date | cst_ds }}'
                    and scan_site_code is not null) a
                     left join jms_dim.dim_network_whole_massage b
                               on a.scan_site_code = b.code) zzc
      group by waybill_no) detail lateral view explode(tab_scan_trace_step(scan_array))  view as exp ) detail
distribute by abs(hash(concat(waybill_no,step))) % 1000;

insert overwrite table jms_dm.dm_tab_scan_piece_detail_dt partition (dt, scan_data_type)
select waybill_no,
       step,
       step_center,
       scan_type,
       scan_time,
       network_code,
       network_name,
       network_type,
       organization_code,
       organization_name,
       data_type,
       pre_site_code,
       pre_site_name,
       next_site_code,
       next_site_name,
       record_id,
       shipment_no,
       shipment_name,
       package_code,
       scan_next_pre_code,
       scan_next_pre_name,
       scan_user,
       scan_user_code,
       scan_user_id,
       scan_user_type,
       send_site,
       send_site_code,
       pcs,
       weight,
       upload_time,
       is_unpacking,
       current_send_package_code,
       current_send_scan_time,
       format_scan_time,
       hours,
       date_time,
       supplier_id,
       supplier_code,
       supplier_name,
       goods_type_id,
       goods_type_code,
       goods_type_name,
       is_parts,
       is_scans,
       sender_city_id,     --寄件城市id
       sender_city_name,   --寄件城市name
       receiver_city_id,   --收件城市id
       receiver_city_name, --收件城市name
       receiver_network_code, --目的网点code
       receiver_network_name, --目的网点
       table_type,         --任务类型；1,干线；2,支线
       bz_type,
       dt,
       scan_data_type
from (select /*+ BROADCAST(institution) */
             waybill_no,
             step,
             step_center,
             info_scan_type                                                                                  scan_type,
             info_scan_time                                                                                  scan_time,
             network_code,
             network_name,
             info_network_type                                                                               network_type,       --网点类型
             nvl(institution.institution_code,flow.institution_code)                                         organization_code,  --中心组织
             nvl(institution.institution_name,flow.institution_name)                                                                           organization_name,  --中心组织
             info_data_type                                                                                  data_type,          --数量类型
             pre_site_code,                                                                                                      --上一站
             pre_site_name,                                                                                                      --上一站
             nvl(next_site_code, info_next_station_code)                                                     next_site_code,     --下一站
             nvl(next_site_name, info_next_station)                                                          next_site_name,     --下一站
             info_record_id                                                                                  record_id,          --记录id
             nvl(info_shipment_no, arrival_shipno_lag)                                                       shipment_no,        --任务号
             b.shipment_name                                                                                 shipment_name,      --任务名
             info_package_code                                                                               package_code,       --包号
             info_next_station_code                                                                          scan_next_pre_code, --扫描记录上下站
             info_next_station                                                                               scan_next_pre_name, --扫描记录上下站
             info_scan_user                                                                                  scan_user,          --扫描员
             info_scan_user_code                                                                             scan_user_code,     --扫描员code
             info_scan_user_id                                                                               scan_user_id,       --扫描员id
             type                                                                                            scan_user_type,     --扫描员类型
             info_send_site                                                                                  send_site,          --寄件网点
             network_info.code                                                                               send_site_code,     --寄件网点
             info_pcs                                                                                        pcs,                --件数
             nvl(info_weight,current_arrival_weight)                                                         weight,             --重量 (发件无重量 从当前到件补)
             info_upload_time                                                                                upload_time,        --上传时间
             0                                                                                               is_unpacking,       --是否拆包
             current_send_package_code,                                                                                          --当前站点发件包号
             current_send_scan_time,                                                                                             --当前站点发件扫描时间
             DATE_FORMAT(info_scan_time, 'yyyy-MM-dd HH:00:00')                                              format_scan_time,   --格式化时间
             hour(info_scan_time)                                                                            hours,
             date(info_scan_time)                                                                            date_time,
             scan_user_info.outsource_company_id                                                             supplier_id,        --供应商id
             scan_user_info.supplier_code                                                                    supplier_code,      --供应商code
             scan_user_info.outsource_company_name                                                           supplier_name,      --供应商name
             waybill.goods_type_id                                                                           goods_type_id,      --货品类型id
             waybill.goods_type_code                                                                         goods_type_code,    --货品类型code
             waybill.goods_type_name                                                                         goods_type_name,    --货品类型name
             if(waybill_no not like 'B%' AND info_package_code IS NULL, 1, 0)                                is_parts,           --是否散件
             if(waybill_no not like 'B%', 1, 0)                                                              is_scans,           --是否扫描件
             waybill.sender_city_id                                                                          sender_city_id,     --寄件城市id
             waybill.sender_city_name                                                                        sender_city_name,   --寄件城市name
             waybill.receiver_city_id                                                                        receiver_city_id,   --收件城市id
             waybill.receiver_city_name                                                                      receiver_city_name, --收件城市name
             waybill.dispatch_network_code                                                                   receiver_network_code, --目的网点
             waybill.dispatch_network_name                                                                   receiver_network_name, --目的网点
             b.table_type                                                                                    table_type,         --任务类型；1,干线；2,支线
             institution.bz_type                                                                                         bz_type, --进港、出港
             date(info_scan_time)                                                                            dt,                 --分区
             info_data_type                                                                                  scan_data_type,     --扫描类型
             row_number() over (partition by waybill_no,step,info_data_type order by flow.modify_time desc ) rnk                 --兜底 避免流向维护重复数据
      from (select *
            from (select *,
                         nvl(send_site_code_lag, arrival_site_code_lag)                        pre_site_code,             --上一站
                         nvl(send_site_name_lag, arrival_site_name_lag)                        pre_site_name,             --上一站
                         nvl(send_site_code_lead, arrival_site_code_lead)                      next_site_code,            --下一站
                         nvl(send_site_name_lead, arrival_site_name_lead)                      next_site_name,            --下一站
                         if(exp_scan_Infos.info_scan_time = '', null,
                            exp_scan_Infos.info_scan_time)                                     info_scan_time,            --扫描时间
                         if(exp_scan_Infos.info_scan_site_code = '', null,
                            exp_scan_Infos.info_scan_site_code)                                info_scan_site_code,       --扫描站点code
                         if(exp_scan_Infos.info_network_type = '', null,
                            exp_scan_Infos.info_network_type)                                  info_network_type,         --网点类型
                         if(exp_scan_Infos.info_scan_site = '', null,
                            exp_scan_Infos.info_scan_site)                                     info_scan_site,            --扫描网点
                         if(exp_scan_Infos.info_data_type = '', null,
                            exp_scan_Infos.info_data_type)                                     info_data_type,            --扫描类型
                         if(exp_scan_Infos.info_record_id = '', null,
                            exp_scan_Infos.info_record_id)                                     info_record_id,            --记录
                         if(exp_scan_Infos.info_shipment_no = '', null,
                            exp_scan_Infos.info_shipment_no)                                   info_shipment_no,          --任务号
                         if(exp_scan_Infos.info_package_code = '', null,
                            exp_scan_Infos.info_package_code)                                  info_package_code,         --包号
                         if(exp_scan_Infos.info_scan_type = '', null,
                            exp_scan_Infos.info_scan_type)                                     info_scan_type,            --扫描类型
                         if(exp_scan_Infos.info_next_station = '', null,
                            exp_scan_Infos.info_next_station)                                  info_next_station,         --发件或者到件下一站
                         if(exp_scan_Infos.info_next_station_code = '', null,
                            exp_scan_Infos.info_next_station_code)                             info_next_station_code,    --发件或者到件下一站
                         if(exp_scan_Infos.info_scan_user = '', null,
                            exp_scan_Infos.info_scan_user)                                     info_scan_user,            --扫描员
                         if(exp_scan_Infos.info_scan_user_code = '', null,
                            exp_scan_Infos.info_scan_user_code)                                info_scan_user_code,       --扫描员code
                         if(exp_scan_Infos.info_scan_user_id = '', null,
                            exp_scan_Infos.info_scan_user_id)                                  info_scan_user_id,         --扫描员id
                         if(exp_scan_Infos.info_send_site = '', null,
                            exp_scan_Infos.info_send_site)                                     info_send_site,            --寄件网点
                         if(exp_scan_Infos.info_pcs = '', null, exp_scan_Infos.info_pcs)       info_pcs,                  --件数
                         if(exp_scan_Infos.info_weight = '', null, exp_scan_Infos.info_weight) info_weight,               --重量
                         if(exp_scan_Infos.info_upload_time = '', null,
                            exp_scan_Infos.info_upload_time)                                   info_upload_time,          --上传时间
                         scan_site_infos_lag,                                                                             --上一站所有信息
                         scan_site_infos_lead,                                                                            --下一站所有信息
                         current_send_info.info_package_code                                   current_send_package_code, --当前发件扫描package_code
                         current_send_info.info_scan_time                                      current_send_scan_time,    --当前发件扫描时间
                         current_arrival_info.info_weight                                      current_arrival_weight,    --当前到件重量
                         current_send_weight                                                                              --当前发件重量(从下一站补充)
                  from (select *,
                               nvl(if(scan_site_infos_lag[0].info_data_type = '1'
                                      , scan_site_infos_lag[0].info_shipment_no,
                                      null),
                                   if( scan_site_infos_lag[1].info_data_type = '1'
                                      , scan_site_infos_lag[1].info_shipment_no,
                                      null)) arrival_shipno_lag,     --当前到件任务号（从上一扫描发件补）
                               case
                                   when scan_site_infos_lag[0].info_data_type = '1'
                                       then scan_site_infos_lag[0].info_scan_site_code
                                   when scan_site_infos_lag[1].info_data_type = '1'
                                       then scan_site_infos_lag[1].info_scan_site_code
                                   else null end                                send_site_code_lag,     --发件扫描code 上一站
                               case
                                   when scan_site_infos_lag[0].info_data_type = '1'
                                       then scan_site_infos_lag[0].info_scan_site
                                   when scan_site_infos_lag[1].info_data_type = '1'
                                       then scan_site_infos_lag[1].info_scan_site
                                   else null end                                send_site_name_lag,     --发件扫描name 上一站
                               case
                                   when scan_site_infos_lag[0].info_data_type = '2'
                                       then scan_site_infos_lag[0].info_scan_site_code
                                   when scan_site_infos_lag[1].info_data_type = '2'
                                       then scan_site_infos_lag[1].info_scan_site_code
                                   else null end                                arrival_site_code_lag,  --到站扫描code 上一站
                               case
                                   when scan_site_infos_lag[0].info_data_type = '2'
                                       then scan_site_infos_lag[0].info_scan_site
                                   when scan_site_infos_lag[1].info_data_type = '2'
                                       then scan_site_infos_lag[1].info_scan_site
                                   else null end                                arrival_site_name_lag,  --到站扫描code 上一站
                               case
                                   when scan_site_infos_lead[0].info_data_type = '1'
                                       then scan_site_infos_lead[0].info_scan_site_code
                                   when scan_site_infos_lead[1].info_data_type = '1'
                                       then scan_site_infos_lead[1].info_scan_site_code
                                   else null end                                send_site_code_lead,    --发件扫描code 下一站
                               case
                                   when scan_site_infos_lead[0].info_data_type = '1'
                                       then scan_site_infos_lead[0].info_scan_site
                                   when scan_site_infos_lead[1].info_data_type = '1'
                                       then scan_site_infos_lead[1].info_scan_site
                                   else null end                                send_site_name_lead,    --发件扫描code 下一站
                               case
                                   when scan_site_infos_lead[0].info_data_type = '2'
                                       then scan_site_infos_lead[0].info_scan_site_code
                                   when scan_site_infos_lead[1].info_data_type = '2'
                                       then scan_site_infos_lead[1].info_scan_site_code
                                   else null end                                arrival_site_code_lead, --到站扫描code 下一站
                               case
                                   when scan_site_infos_lead[0].info_data_type = '2'
                                       then scan_site_infos_lead[0].info_scan_site
                                   when scan_site_infos_lead[1].info_data_type = '2'
                                       then scan_site_infos_lead[1].info_scan_site
                                   else null end                                arrival_site_name_lead, --到站扫描name 下一站
                               case
                                   when scan_site_infos[0].info_data_type = '1' then scan_site_infos[0]
                                   when scan_site_infos[1].info_data_type = '1' then scan_site_infos[1]
                                   else null end                                current_send_info,       --当前发件扫描信息
                               case
                                   when scan_site_infos[0].info_data_type = '2' then scan_site_infos[0]
                                   when scan_site_infos[1].info_data_type = '2' then scan_site_infos[1]
                                   else null end                                current_arrival_info,     --当前到件扫描信息
                               nvl(if(scan_site_infos_lead[0].info_weight = '', null,
                                      scan_site_infos_lead[0].info_weight),
                                   if(scan_site_infos_lead[1].info_weight = '', null,
                                      scan_site_infos_lead[1].info_weight)) current_send_weight           --当前发件体积（从下一扫描到件补）
                        from (SELECT *,
                                     LAG(scan_site_infos, 1, null)
                                         over (partition by waybill_no order by step)  scan_site_infos_lag, --上一条扫描信息
                                     lead(scan_site_infos, 1, null)
                                          over (partition by waybill_no order by step) scan_site_infos_lead --下一条扫描信息
                              FROM (select waybill_no,
                                           step,
                                           stepcenter step_center,    --中心步数
                                           network_code,
                                           network_name,
                                           collect_set(
                                                   named_struct(
                                                           'info_scan_time',
                                                           infos[0],
                                                           'info_scan_site_code',
                                                           infos[1],
                                                           'info_network_type',
                                                           infos[2],
                                                           'info_scan_site',
                                                           infos[3],
                                                           'info_data_type',
                                                           infos[4],
                                                           'info_record_id',
                                                           infos[5],
                                                           'info_shipment_no',
                                                           infos[6],
                                                           'info_package_code',
                                                           infos[7],
                                                           'info_scan_type',
                                                           infos[8],
                                                           'info_next_station',
                                                           infos[9],
                                                           'info_next_station_code',
                                                           infos[10],
                                                           'info_scan_user',
                                                           infos[11],
                                                           'info_scan_user_code',
                                                           infos[12],
                                                           'info_scan_user_id',
                                                           infos[13],
                                                           'info_send_site',
                                                           infos[14],
                                                           'info_pcs',
                                                           infos[15],
                                                           'info_weight',
                                                           infos[16],
                                                           'info_upload_time',
                                                           infos[17]
                                                       ) --转化成struct
                                               )      scan_site_infos --到发数据合成array
                                    from (select waybill_no,
                                                 step,
                                                 stepcenter,
                                                 network_code,
                                                 network_name,
                                                 split(scan_Infos, '@') infos --切分数据
                                          from jms_tmp.tmp_tab_scan_step_detail) info_ls
                                    group by waybill_no,
                                             step,
                                             stepcenter,
                                             network_code,
                                             network_name) DETAIL) pre_detail) detail lateral view explode(scan_site_infos) exp as exp_scan_Infos) filter_detail
            where date(info_scan_time) between date_add('{{ execution_date | cst_ds }}', -2 + 1) and date_add('{{ execution_date | cst_ds }}', -1 + 1)
              and info_network_type = 4) etl_detail
               left join
           (select shipment_no, shipment_name , table_type
            from jms_dwd.dwd_tmsnew_shipment_union_base_dt
            where dt between date_add('{{ execution_date | cst_ds }}', -15 + 1) and date_add('{{ execution_date | cst_ds }}', -1 + 5)) b
           on coalesce(etl_detail.info_shipment_no, etl_detail.arrival_shipno_lag, etl_detail.waybill_no) =
              b.shipment_no --补充任务名
               left join jms_dim.dim_network_whole_massage network_info
                         on etl_detail.info_send_site = network_info.name  --补充寄件网点名
               left join (select waybill_no billcode,
                                 pick_network_code,
                                 pick_network_name,
                                 sender_city_id,
                                 sender_city_name,
                                 receiver_city_id,
                                 receiver_city_name,
                                 dispatch_network_code,
                                 dispatch_network_name,
                                 goods_type_id,
                                 goods_type_code,
                                 goods_type_name
                          from jms_dwd.dwd_yl_oms_oms_waybill_incre_dt
                          where dt between date_add('{{ execution_date | cst_ds }}', -3 + 1) and date_add('{{ execution_date | cst_ds }}', -1 + 1)) waybill
                         on waybill.billcode = etl_detail.waybill_no  --补充物品类型
               left join (
                          select * from (
                                 select
                                   center_code,
                                   center_name,
                                   scan_type + 1 scan_type,
                                   direction_code direction_code_array,
                                   direction_name,
                                   institution_code,
                                   institution_name,
                                   assert_start,
                                   assert_end,
                                   modify_time,
                                   flow_code,
                                   row_number() over(partition by center_code,scan_type,flow_code,assert_start,assert_end order by modify_time desc) rn
                                 from
                                   jms_dim.dim_ep_flow_direction_base
                                 lateral view explode(split(direction_code,',')) t as flow_code
                                 ) detail
                                 where rn = 1
                          ) flow   --补充流向数据
                         on flow.center_code = etl_detail.info_scan_site_code
                         and flow.scan_type = etl_detail.info_data_type
                         and direction_code_array = if(etl_detail.info_data_type = 1, etl_detail.next_site_code,etl_detail.pre_site_code)
                         and flow.assert_start <= etl_detail.info_scan_time
                         and flow.assert_end >= etl_detail.info_scan_time
               left join (select code,
                                 outsource_company_id,
                                 outsource_company_name,
                                 supplier_code,
                                 type,
                                 modify_time,
                                 row_number() over (partition by code order by modify_time desc ) rnk
                          from jms_dim.dim_ep_center_person_base) scan_user_info  --补充人员信息
                         on scan_user_info.rnk = 1
                             and etl_detail.info_scan_user_code = scan_user_info.code
                left join (
                    select center_code,
                           employ_code,
                           sign_in_time,
                           nvl(sign_out_time,work_end_time) sign_out_time,
                           bz_type,
                           dt,
                           institution_code,
                           institution_name
                    from (
                           select a.center_code,
                                  b.employ_code,
                                  a.work_start_time,
                                  a.work_end_time,
                                  b.sign_in_time,
                                  b.sign_out_time,
                                  post.bz_type,
                                  date(a.create_day) dt,
                                  a.institution_code,
                                  a.institution_name,
                                  row_number() over (partition by a.center_code,b.employ_code,b.sign_in_time,b.sign_out_time order by a.create_day desc) rn
                           from jms_dwd.dwd_ep_create_institution_base  a
                           left join jms_dwd.dwd_ep_create_institution_detail_base b
                           on a.id = b.create_institution_id left join
                           jms_dim.dim_ep_center_institutiona_post_dt post
                           on post.code = a.institution_code
                           and post.site_code = a.center_code
                           and post.post_code = b.post_code
                           and post.dt between date_add('{{ execution_date | cst_ds }}', -3 + 1) and date_add('{{ execution_date | cst_ds }}', -1 + 1)
                           and post.dt = date(a.create_day)
                           where date(a.create_day) between date_add('{{ execution_date | cst_ds }}', -3 + 1) and date_add('{{ execution_date | cst_ds }}', -1 + 1)
                         ) detail where rn = 1
                ) institution on etl_detail.info_scan_time >= institution.sign_in_time
                           and   etl_detail.info_scan_time <= institution.sign_out_time
                           and    etl_detail.info_scan_user_code = institution.employ_code
                           and    etl_detail.network_code= institution.center_code
                ) detail
where rnk = 1
distribute by dt
, scan_data_type
, abs(hash(waybill_no))%120;